package com.yyx.jupiter.service.recommend.rerank;


import com.alibaba.fastjson2.JSONObject;
import com.yyx.jupiter.component.RedisClient;
import com.yyx.jupiter.conf.PropertyConfig;
import com.yyx.jupiter.entity.RecomReq;
import com.yyx.jupiter.entity.RecomVideo;
import com.yyx.jupiter.utils.TimeUtils;
import jakarta.annotation.Resource;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.math3.distribution.BetaDistribution;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Stream;

@Slf4j
@Service
public class ThompsonSampleRerank implements Rerankable<RecomVideo> {

    @Resource
    PropertyConfig propertyConfig;

    @Resource
    RedisClient redisClient;

    @Override
    public List<RecomVideo> rerank(List<RecomVideo> recomItems, RecomReq recomReq) throws Exception {

        long startTime = System.currentTimeMillis();

        // 请求体参数
        String requestId = recomReq.getRequestId();
        String userId = recomReq.getUserId();
        String strategy = recomReq.getStrategy();

        // 配置参数
        JSONObject strategyJson = propertyConfig.getStrategy().getJSONObject(strategy);
        boolean logPrintSwitch = MapUtils.getBooleanValue(strategyJson, "logPrintSwitch", false);
        float timeFactor = MapUtils.getFloatValue(strategyJson, "timeFactor", 10e5F);
        float hobbyFactor = MapUtils.getFloatValue(strategyJson, "hobbyFactor", 0);

        // 获取用户喜好
        String hobbies = redisClient.get(userId, String.class);
        hobbies = hobbies == null ? "" : hobbies;

        // 贝叶斯平滑
        List<RecomVideo> newRecomVideos = recomItems.stream()
                .filter(i -> i.getRecallStrategy().equals("newVideo")).toList();
        Map<String, RecomVideoStat> recomVideoStats = redisClient
                .multiGet(newRecomVideos.stream().map(RecomVideo::getVideoId).toList(), RecomVideoStat.class);

        // 汤普森采样
        for (RecomVideo newRecomVideo : newRecomVideos) {
            String redisKey = "yyx:video_stat:" + newRecomVideo.getVideoId();
            int exposeCnt = 2;
            int clickCnt = 0;
            if (recomVideoStats.containsKey(redisKey)) {
                exposeCnt = recomVideoStats.get(redisKey).getExpose().intValue();
                clickCnt = Stream.of(
                        recomVideoStats.get(redisKey).getLike(),
                        recomVideoStats.get(redisKey).getCollect(),
                        recomVideoStats.get(redisKey).getComment(),
                        recomVideoStats.get(redisKey).getFinish(),
                        recomVideoStats.get(redisKey).getShare(),
                        recomVideoStats.get(redisKey).getJump()
                ).max(Long::compare).get().intValue();
            }
            // 处理异常情况
            clickCnt = clickCnt >= exposeCnt ? exposeCnt - 1 : clickCnt;
            BetaDistribution betaDistribution = new BetaDistribution(clickCnt, exposeCnt - clickCnt);
            float sampleScore = Float.parseFloat(String.valueOf(betaDistribution.sample()));
            newRecomVideo.setScore(sampleScore);
        }

        // 计算综合得分
        for (RecomVideo newRecomVideo : newRecomVideos) {
            boolean isHobby = hobbies.contains(newRecomVideo.getSecondCategory());
            String createTime = newRecomVideo.getCreateTime();
            int dayTillNow = TimeUtils.dateDiff(createTime.substring(0, 10), TimeUtils.getNowTime("yyyy-MM-dd"));
            float hobbyScore = hobbyFactor * (isHobby ? 1 : 0);
            float timeScore = Float.parseFloat(String.valueOf(Math.log(Math.abs(dayTillNow)) / Math.log(timeFactor)));
            float score = 100 + newRecomVideo.getScore() + hobbyScore - timeScore;
            newRecomVideo.setScore(score);
        }


        // 重新排序
        List<String> newRecomVideoIds = newRecomVideos.stream().map(RecomVideo::getVideoId).toList();
        List<RecomVideo> otherRecomList = recomItems.stream()
                .filter(i -> !newRecomVideoIds.contains(i.getVideoId())).toList();
        List<RecomVideo> resultRecomList = new ArrayList<>();
        resultRecomList.addAll(newRecomVideos);
        resultRecomList.addAll(otherRecomList);

        if (logPrintSwitch) {
            log.info("requestId:{}, thompson sampling elapsed: {} ms", requestId, (System.currentTimeMillis() - startTime));
        }
        return resultRecomList.stream().sorted(Comparator.comparing(RecomVideo::getScore).reversed()).toList();
    }

    @Data
    public static class RecomVideoStat {
        private String videoId;
        private Long expose;
        private Long like;
        private Long collect;
        private Long comment;
        private Long finish;
        private Long share;
        private Long jump;
    }
}
